package com.solver.demo.rscoket.v2.client;

import com.solver.demo.rscoket.v2.api.ChannelMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.UUID;

/**
 * @author：fangn
 * @since：2025/4/2
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class V2RSocketClient {
    private final RSocketRequester requester;

    public Flux<String> sendGroupedMessages(List<String> contents, Duration timeout) {
        String autoGroupId = "group-" + UUID.randomUUID();

        Flux<ChannelMessage> requests = Flux.fromIterable(contents)
                .delayElements(Duration.ofMillis(100))
                .map(content -> ChannelMessage.builder()
                        .groupId(autoGroupId)
                        .content(content)
                        .timestamp(LocalDateTime.now())
                        .build());

        return requester.route("channelV2")
                .data(requests)
                .retrieveFlux(String.class)
                .timeout(timeout)
                .doOnSubscribe(s -> log.info("Sending {} messages in group {}", contents.size(), autoGroupId))
                .doOnNext(response -> log.info("Received response: {}", response)) // 添加响应日志输出
                .doOnError(e -> log.error("Channel error: {}", e.getMessage()));
    }

}

